#ifndef BLOCK_H
#define BLOCK_H

#include <muduo/base/Types.h>
#include <muduo/base/Logging.h>

#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>

#include <sdfs/common/Pool.h>
#include <sdfs/common/File.h>
#include <sdfs/common/ToolKit.h>

const long THREADHOLD = 50;

using namespace muduo;
using namespace boost;
using namespace sdfs::common;

namespace sdfs
{
namespace storage
{

class Piece: public enable_shared_from_this<Piece>
{
public:
	Piece(int idx, long offset, long length):
		index_(idx), offset_(offset), length_(length), pointer_(offset)
	{

	}

	~Piece()
	{

	}


	shared_ptr<Piece> split(long /* length of right */ length)
	{
		assert(length_ >= length);
		long len = length_ - length;
		if(len != 0)
		{
			long offset2 = offset_ + len;
			length_ = len;
			shared_ptr<Piece> piece(new Piece(index_, offset2, length));
			return piece;
		}
		return shared_from_this();
	}

	bool connected(const shared_ptr<Piece> p)
	{
		return  ( p->offset_  == offset_ + length_ ) ? true : false;
	}

	void setOffset(long offset)
	{
		offset_ = offset;
	}

	void setLength(long length)
	{
		length_ = length;
	}

	void enlargeLength(long length)
	{
		length_ += length;
	}

	bool isFeet(long length)
	{
		return length_ >= length;
	}

	bool operator == (const Piece& piece)
	{
		return offset_ == piece.offset_ && length_ == piece.length_;
	}

	bool operator < (const Piece& piece)
	{
		if (piece.offset_ < offset_)
		{
			return false;
		}
		else if(piece.offset_ > offset_)
		{
			return true;
		}
		return piece.length_ > length_ ? true : false;
	}

	bool operator > (const Piece& piece)
	{
		if (piece.offset_ > offset_)
		{
			return false;
		}
		else if(piece.offset_ < offset_)
		{
			return true;
		}
		return piece.length_ < length_ ? true : false; 
	}

	long getOffset()
	{
		return offset_;
	}

	long getLength()
	{
		return length_;
	}

	int getIndex()
	{
		return index_;
	}

	long getPointer()
	{
		return pointer_;
	}

	void slipPointer(long offset)
	{
		pointer_ += offset;
		assert(pointer_ <= offset_+length_);
	}

	bool isFinished()
	{
		return pointer_ == offset_+length_ ? true:false
	}

	void rewind()
	{
		pointer_ = offset_;
	}

private:
	int index_;
	long offset_;
	long length_;
	long pointer_;
};

typedef shared_ptr<Piece> PiecePtr;

class Block
{
public:
	Block(int index, const muduo::string& fullname, int poolsize, long maxsize):
		index_(index),
		fullname_(fullname),
		poolsize_(poolsize),
		lastPos_(0),
		readFiles_( new Pool<ReadOnlyFile>(poolsize_)),
		writeFiles_( new Pool<WriteOnlyFile>(poolsize_))
	{
		// blocksize_ = sdfs::ToolKit::getFileSize(fullname_.c_str());
		LOG_DEBUG << "create READONLYFILES";
		posUpdater_.reset(new WriteOnlyFile(fullname_, maxsize));
		lastPos_ = posUpdater_->lastPos();
		for (int i = 0; i < poolsize; ++i)
		{
			WriteOnlyFile* writer = new WriteOnlyFile(fullname_, maxsize);
			writer->registerToPool(writeFiles_.get());

			ReadOnlyFile* reader = new ReadOnlyFile(fullname_);
			reader->registerToPool(readFiles_.get());
			
		}
	}
	~Block()
	{
	}

	void updateLastPos(long pos)
	{
		lastPos_ = pos;
		posUpdater_->updateLastPos(pos);
	}

	long getLastPos()
	{
		return lastPos_;
	}

	ReadOnlyFilePtr getReader()
	{
		ReadOnlyFilePtr reader;
		if(!readFiles_->avaliable())
		{
			LOG_INFO << "not enough reader, add a reader to pool";
			ReadOnlyFile* file = new ReadOnlyFile(fullname_);
			file->registerToPool(readFiles_.get());
		}
		reader = readFiles_->getData();
		return reader;
	}

	WriteOnlyFilePtr getWriter()
	{
		//MutexLockGuard guard(writerMutex_);
		//assert(writing_ == false);
		//writing_ = true;
		return writeFiles_->getData();
	}

	int getIndex()
	{
		return index_;
	}

	/* data */
private:
	int index_;
	string fullname_;
	int poolsize_;
	long lastPos_;

	shared_ptr< Pool<ReadOnlyFile> > readFiles_;
	// muduo::MutexLock writerMutex_;
	shared_ptr< Pool<WriteOnlyFile> > writeFiles_;

	WriteOnlyFilePtr posUpdater_;

};

typedef shared_ptr<Block> BlockPtr;


}
}

#endif